package com.chengqs.dfs.namenode.server;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

/**
 * <B>主类名称：</B>FSNameSystem<BR>
 * <B>概要说明：</B>负责管理元数据的核心组件<BR>
 *
 * @author: chengqs
 * @create: 2025-03-30 03:48:59
 **/
@Slf4j
public class FSNameSystem {
    /**
     * 负责管理内存文件目录树的组件
     */
    private FSDirectory directory;
    /**
     * 负责管理edits log写入磁盘的组件
     */
    private FSEditLog editlog;

    // 最近一次checkpoint更新到的txid
    private long checkpointTxid = 0;

    public FSNameSystem() {
        this.directory = new FSDirectory();
        this.editlog = new FSEditLog(this);
        recoverNamespace();
    }

    /**
     * 创建目录
     * @param path 目录路径
     * @return 是否成功
     */
    public Boolean mkdir(String path) throws Exception {
        this.directory.mkdir(path);
        this.editlog.logEdit(EditLogFactory.mkdir(path));
        return true;
    }

    /**
     * 创建文件
     * @param filename 文件名，包含所在的绝对路径，/products/img001.jpg
     * @return
     * @throws Exception
     */
    public Boolean create(String filename) throws Exception {
        if (!directory.create(filename)) {
            return false;
        }
        // 在这里写一条editlog
        this.editlog.logEdit(EditLogFactory.create(filename));
        return true;
    }

    // 强制把内存里的editslog刷入磁盘中
    public void flush() {
        this.editlog.flush();
    }

    // 获取一个FSEditLog组件
    public FSEditLog getEditLog() {
        return editlog;
    }

    public void setCheckpointTxid(long txid) {
        log.info("接收到checkpoint txid：{}", txid);
        this.checkpointTxid = txid;
    }

    public long getCheckpointTxid() {
        return checkpointTxid;
    }

    /**
     * 将checkpointTxid保存到磁盘上去
     */
    public void saveCheckpointTxid() {
        String path = "D:\\java\\儒猿课堂\\儒猿海量小文件分布式存储项目\\checkpoint-txid.meta";

        RandomAccessFile raf = null;
        FileOutputStream out = null;
        FileChannel channel = null;

        try {
            File file = new File(path);
            if (file.exists()) {
                file.delete();
            }

            ByteBuffer buffer = ByteBuffer.wrap(String.valueOf(checkpointTxid).getBytes());

            raf = new RandomAccessFile(path, "rw");
            out = new FileOutputStream(raf.getFD());
            channel = out.getChannel();

            channel.write(buffer);
            channel.force(false); // 强制把数据刷到磁盘上
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (out != null) {
                    out.close();
                }
                if (raf != null) {
                    raf.close();
                }
                if (channel != null) {
                    channel.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 恢复元数据
     */
    public void recoverNamespace() {
        try {
            loadFSImage();
            loadCheckpointTxid();
            loadEditLog();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 加载fsImage文件到内存里进行恢复
     */
    private void loadFSImage() throws Exception {
        FileInputStream in = null;
        FileChannel channel = null;
        try {
            String path = "D:\\java\\儒猿课堂\\儒猿海量小文件分布式存储项目\\fsImage.meta";
            File file = new File(path);
            if (!file.exists()) {
                log.warn("fsimage文件当前不存在，不进行恢复......");
                return;
            }

            in = new FileInputStream(path);
            channel = in.getChannel();

            ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024); // 这个参数是可以动态调节的
            // 每次接收到一个fsimage文件的时候记录一下它的大小，持久化到磁盘上
            // 每次重启就分配对应的空间大小就可以了
            int count = channel.read(buffer);
            buffer.flip();
            String fsImageJson = new String(buffer.array(), 0, count);

            log.info("恢复fsImage文件中的数据：{}", fsImageJson);

            FSDirectory.INode dirTree = JSONObject.parseObject(fsImageJson, FSDirectory.INode.class);
            this.directory.setDirTree(dirTree);
        } finally {
            try {
                if (in != null) {
                    in.close();
                }
                if (channel != null) {
                    channel.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 加载和回放editlog
     * @throws Exception
     */
    private void loadEditLog() throws Exception {
        File dir = new File("D:\\java\\儒猿课堂\\儒猿海量小文件分布式存储项目\\");

        List<File> files = new ArrayList<>();
        for (File file : dir.listFiles()) {
            if (file.getName().contains("edits")) {
                files.add(file);
            }
        }

        Collections.sort(files, (o1, o2) -> {
            int o1StartTxid = Integer.valueOf(o1.getName().split("-")[1]);
            int o2StartTxid = Integer.valueOf(o2.getName().split("-")[1]);
            return o1StartTxid - o2StartTxid;
        });

        if (files == null || files.size() == 0) {
            log.warn("当前目录下没有editlog文件，不进行恢复......");
            return;
        }
        for (File file : files) {
            if (file.getName().contains("edits")) {
                String[] splitName = file.getName().split("-");
                long startTxid = Long.parseLong(splitName[1]);
                long endTxid = Long.parseLong(splitName[2].split("[.]")[0]);

                // 如果checkpointTxid之后的那些editlog都要加载出来
                if (endTxid > checkpointTxid) {
                    String currentEditsLogFile = "D:\\java\\儒猿课堂\\儒猿海量小文件分布式存储项目\\edits-" +
                            startTxid + "-" + endTxid + ".log";
                    List<String> editsLogs = Files.readAllLines(Paths.get(currentEditsLogFile), StandardCharsets.UTF_8);

                    for (String editsLog : editsLogs) {
                        JSONObject editLog = JSONObject.parseObject(editsLog);
                        long txid = editLog.getLongValue("txid");

                        if (txid > checkpointTxid) {
                            log.info("准备回放edit log：{}", editsLog);
                            // 回放到内存里去
                            String op = editLog.getString("OP");

                            if (op.equals("MKDIR")) {
                                String path = editLog.getString("PATH");
                                try {
                                    directory.mkdir(path);
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            } else if (op.equals("CREATE")) {
                                String path = editLog.getString("PATH");
                                try {
                                    directory.create(path);
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    private void loadCheckpointTxid() throws Exception {
        FileInputStream in = null;
        FileChannel channel = null;
        try {
            String path = "D:\\java\\儒猿课堂\\儒猿海量小文件分布式存储项目\\checkpoint-txid.meta";
            File file = new File(path);
            if (!file.exists()) {
                log.warn("checkpointTxid文件当前不存在，不进行恢复......");
                return;
            }

            in = new FileInputStream(path);
            channel = in.getChannel();

            ByteBuffer buffer = ByteBuffer.allocate(1024); // 这个参数是可以动态调节的
            // 每次接收到一个fsimage文件的时候记录一下它的大小，持久化到磁盘上
            // 每次重启就分配对应的空间大小就可以了
            int count = channel.read(buffer);
            buffer.flip();
            Long checkpointTxid = Long.valueOf(new String(buffer.array(), 0, count));

            log.info("恢复checkpointTxid：{}", checkpointTxid);

            this.checkpointTxid = checkpointTxid;
        } finally {
            try {
                if (in != null) {
                    in.close();
                }
                if (channel != null) {
                    channel.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
